package com.lxn.app

import com.alibaba.fastjson.{JSON, JSONObject}
import com.lxn.bean.{DauInfo, PageLog}
import com.lxn.until.{MYKafkaUtils, MyBeanUtils, MyEsUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.{Jedis, Pipeline}

import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util.Date
import scala.collection.mutable.ListBuffer

object DWDDauApp {

  def main(args: Array[String]): Unit = {

    //每次启动，还原数据状态
    revertStatus()

    //还是一样的流程 准备实时环境 读取偏移量 创建消费 提取偏移量结束点 转换数据格式 开始处理数据
    val sparkConf = new SparkConf().setAppName("dwd_dau_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val topicName : String = "DWD_PAGE_LOG"
    val groupId : String = "DWD_DAU_GROUP"
    //读取偏移量
    val offsets : Map[TopicPartition,Long] = MyOffsetUtils.readOffset(groupId, topicName)
    //创建消费
    var KafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if(offsets!=null && offsets.nonEmpty){
      KafkaDStream = MYKafkaUtils.getConsumer(ssc, topicName, groupId ,offsets)
    }else{
      KafkaDStream = MYKafkaUtils.getConsumer(ssc, topicName, groupId)
    }
    //提取结束偏移量
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream :DStream[ConsumerRecord[String,String]] = KafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )
    //转换数据结构
    val pageLogDStream = offsetRangesDStream.map(
      consumerRecord => {
        //获取值
        val value = consumerRecord.value()
        //转换数据结构，我们转换成之前的专用类型
        val pageLog = JSON.parseObject(value, classOf[PageLog])
        pageLog
      }
    )
//    pageLogDStream.print(100)
    pageLogDStream.foreachRDD(
      rdd => {
        println("自我审查前: "+rdd.count())
      }
    )

    //数据去重
    val filterDStream = pageLogDStream.filter(
      PageLog => PageLog.last_page_id == null
    )
    filterDStream.foreachRDD(
      rdd => {
        println("自我审查后: "+rdd.count())
        println("-------------------------------")
      }
    )
    //上面的去重只可以过滤出登录次数，但一个人可能一天多次访问页面，所以我们需要对一天内相同的mid或者uid进行去重
//    filterDStream.filter()这个也能够去重，但是，这个会对每一条数据进行处理，所以会频繁导致jedis连接对象的开关
    val redisFilterDStream = filterDStream.mapPartitions( //使用这个虽然减少了jedis连接的次数，但是数据需要我们自己保存
      pageLogItr => {
        //由于pageLogItr是一个迭代器，所以直接使用size查看会是数据没有，相当于迭代过了，所以我们转成list可以迭代多次
        val pageLogList: List[PageLog] = pageLogItr.toList
        println("第三方审查前: "+pageLogList.size)
        //我们使用第三方工具redis记录用户数据，然后判断是否已经登陆过，没有就加入存储
        //定义一个保存数据的流
        //我们指定key为天
        val pageLogs: ListBuffer[PageLog] = ListBuffer[PageLog]()
        val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
        //在循环外面创建连接对象
        val jedis = MyRedisUtils.getRedisFromPool()
        for (pageLog <- pageLogList) {
          //对于每一个对象进行处理
          val mid: String = pageLog.mid
          val ts: Long = pageLog.ts
          val date: Date = new Date(ts)
          val dateStr: String = sdf.format(date)
          //拼接key
          val redisKey: String = s"DAU_$dateStr"
          //根据key进行查找
          //按照一般思路去实现方法  这种方法在单线程的任务适合，但是分布式任务会有多个并发度，所以同样的数据可能同时写入成功，所以不适用
          //获取所有的key值，看看是否包含传入的值
          //list
          /*val midList: util.List[String] = jedis.lrange(redisKey, 0, -1)
          if(!midList.contains(mid)){
            //不包含就说明是新数据
            jedis.lpush(redisKey,mid)
            //保存这条数据
            pageLogs.append(pageLog)
          }
          //set
          val midSet : util.Set[String] = jedis.smembers(redisKey)
          if(!midSet.contains(mid)){
            jedis.sadd(redisKey,mid)
            pageLogs.append(pageLog)
          }*/

          //如果我们更加了解这个set集合，我们知道在set集合里面，插入数据成功，就会是1，失败就是0，而redis是单线程操作，所以这个事务具有原子性，就可以直接插入，保证数据不重复
          val isNew: Long = jedis.sadd(redisKey, mid)
          if (isNew == 1L) {
            //插入成功，保存数据
            pageLogs.append(pageLog)
          }
        }
        jedis.close()
        println("第三方审查之后: "+pageLogs.size)
        pageLogs.iterator
      }

    )
    //由于这个mapPartitions是转换算子，所以想要看到结果还得调用行动算子
//    redisFilterDStream.print()

    //维度关联
    //为什么用mapPartitions 还是因为我们是分布式环境，有多个分区，所以必须对每个分区的数据进行处理，如果单线程，用foreachRDD就可以了
    val dauDStream = redisFilterDStream.mapPartitions(
      //上面一级是流对象，它包含了一段时间的pageLog对象
      //可以通过迭代器对每一个段流进行处理
      pageLogItr => {
        //for循环是针对每一段流里面的个体，每一个pageLog对象
        //创建连接对象
        val jedis: Jedis = MyRedisUtils.getRedisFromPool()
        //创建一个用于保存数据的
        val dauInfos: ListBuffer[DauInfo] = ListBuffer[DauInfo]()
        //自定时间格式
        val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        for (pageLog <- pageLogItr) {
          //创建一个类型用来存储所有有关日活的信息，后续分析需要用到的字段
          val dauInfo: DauInfo = new DauInfo()
          //然后我们需要将所需要的字段赋值给这个全新的类型
          MyBeanUtils.copyProperties(pageLog, dauInfo)

          //补充维度数据
          //补充用户表维度  因为我们之前在上一个app中将业务数据分为了事实表和维度表，并将他们保存到redis中，所以我们可以根据key获取值
          //获取id,然后拼接key，获取需要拼接的字段
          val uid = pageLog.user_id
          val redisKey: String = s"DIM:USER_INFO:$uid"
          //获取全部数据
          val userInfoJson = jedis.get(redisKey)
          //方便提取数据，我们将字符串转换为对象
          val userInfoJsonObj = JSON.parseObject(userInfoJson)
          //获取需要的字段
          val gender: String = userInfoJsonObj.getString("gender")
          val birthday: String = userInfoJsonObj.getString("birthday")
          //我们需要的是年龄，所以我们需要进行数据转换
          val birthdayLd: LocalDate = LocalDate.parse(birthday)
          //获取当前时间
          val nowDate: LocalDate = LocalDate.now()
          val period = Period.between(birthdayLd, nowDate)
          val age: Int = period.getYears




          //好了，现在可以添加维度了
          dauInfo.user_gender = gender
          dauInfo.user_age = age.toString




          //然后是地区维度数据
          val provinceId: String = dauInfo.province_id
          val redisProvinceKey: String = s"DIM:BASE_PROVINCE:$provinceId"
          val provinceJson: String = jedis.get(redisProvinceKey)
          val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
          val provinceName: String = provinceJsonObj.getString("name")
          val provinceAreaCode: String = provinceJsonObj.getString("area_code")
          val provinceIosCode: String = provinceJsonObj.getString("ios_code")
          val provinceIos3166: String = provinceJsonObj.getString("ios_3166_2")

          dauInfo.province_iso_code = provinceIosCode
          dauInfo.province_3166_2 = provinceIos3166
          dauInfo.province_area_code = provinceAreaCode
          dauInfo.province_name = provinceName

          //日期补充，观察我们传入的数据，包括pageLoge的所传过去且接收的数据，和用户维度数据，地区维度数据，现在只差两个日期
          //将ts时间戳转化为格式化时间，然后拆分字符串获取需要的时间
          //将时间戳转换为格式化时间
          val date: Date = new Date(pageLog.ts)
          //转换为字符串
          val dateStr: String = sdf.format(date)
          //拆分字符串，获取哪一天
          val ti: Array[String] = dateStr.split(" ")
          //继续拆分，获取小时
          val tp: String = ti(1).split(":")(0)

          //合并数据
          dauInfo.dt = ti(0)
          dauInfo.hr = tp
          //保存数据
          dauInfos.append(dauInfo)
        }
        jedis.close()
        dauInfos.iterator
      }

    )

    //将数据写入es
    //我们要求写入的格式是List[(String,Anyef)]
    dauDStream.foreachRDD(
      rdd =>{
        //所以我们先要将数据进行格式转换
        //因为数据在多个分区，所以我们使用这个算子
        rdd.foreachPartition(
          dauInfoIter =>{
            //对每一个迭代器进行转换
            val docs: List[(String, DauInfo)]
            = dauInfoIter.map(dauInfo => (dauInfo.mid, dauInfo)).toList
            //进行判断，数据不为空才可以进行操作
            if(docs.size>0){
              //拼接索引名
              //根据每一天的日期创建索引，生产环境直接使用当前日期，但是现在测试需要获取时间
              val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd")
              //从第一条数据中获取时间
              val head: (String, DauInfo) = docs.head
              val ts :Long = head._2.ts
              val dateStr :String = sdf.format(new Date(ts))
              sdf.format(new Date())
              val indexName : String = s"gmall_dau_info$dateStr"
              //现在只需要插入数据就好了
              MyEsUtils.bulkSave(indexName, docs)
            }
          }
        )
        //提交偏移量
        MyOffsetUtils.saveOffset(groupId,topicName,offsetRanges)
      }
    )

    dauDStream.print(100)

    ssc.start()
    ssc.awaitTermination()

  }

  //还原状态，由于使用第三方工具保存数据，可能存在数据不同步问题，所以我们希望让es的数据区替换redis保存的数据
  //我们需要
  def revertStatus(): Unit = {
    //由于存在数据上的不同，所以我们得以我们真实存储的数据为准，我们让es的数据更新redis中的数据
    //但是我们在测试环境，这个东西只能重置当天的数据
    val date: LocalDate = LocalDate.now()
    //我们先获取索引和字段，因为是根据字段进行查询的
    val indexName : String = s"gmall_dau_info$date"
    val fieldName : String = "mid"
    //根据索引和字段在es中查询所有mid
    val mids: List[String] = MyEsUtils.searchField(indexName, fieldName)
    //先删除redis里面保留的所有索引
    val redisKey : String = s"DAU_$date"
    val jedis = MyRedisUtils.getRedisFromPool()
    jedis.del(redisKey)
    //然后判断数据，就是说es里面有值才能进行更新替换
    if(mids!=null && mids.size>0){
      //现在我们需要使用管道操作，一次性将数据更新
      val pipeline: Pipeline = jedis.pipelined()
      //然后将每一个mid放入管道中，然后一起发送到redis执行
      for(mid <- mids){
        pipeline.sadd(redisKey,mid)  //不会在redis执行
      }
      pipeline.sync()  //一起在redis执行更新
    }
  }

}
